package com.example.provider;

import com.example.provider.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.jupiter.api.Test;

/**
 * @description: TODO
 * @author: uu
 * @date: 2023/12/14 21:10
 * @version: 1.0
 */

public class Tag {
    @Test
    public void tagProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("tagProducer");
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        producer.start();
        Message msg1 = new Message("tagTopic", "vip1", "vip1的消息".getBytes());
        Message msg2 = new Message("tagTopic", "vip2", "vip2的消息".getBytes());
        SendResult s1res = producer.send(msg1);
        SendResult s2res = producer.send(msg2);


    }

    @Test
    public void vip1TagConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("vip1Consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        consumer.subscribe("tagTopic", "vip1");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) ->{
            System.out.println("vip1 消费者：" + new String(msgs.get(0).getBody()));

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
    @Test
    public void vip2TagConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("vip2Consumer");
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        consumer.subscribe("tagTopic", "vip1 || vip2");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) ->{
            System.out.println("vip2 消费者：" + new String(msgs.get(0).getBody()));

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
